package net.hn.hnms.biz.upkafka.service;


import lombok.extern.slf4j.Slf4j;
import net.hn.hnms.biz.upkafka.config.KafkaProperties;
import net.hn.hnms.biz.upkafka.config.SpringContextUtil;
import net.hn.hnms.biz.upkafka.domain.CoalflowNowPo;
import net.hn.hnms.biz.upkafka.domain.SynergiaNowPo;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 综采系统
 */
@Slf4j
public class ZCProducer extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    public final KafkaProducer<String, String> producer;

    private final String topic;
    private final String mineCode;

    public final Boolean isAsync;
    public ZCProducer(String topicName, Boolean asyncEnable,String mineCode) {
        Properties props = Producer.initProperties(topicName,mineCode);
        producer = new KafkaProducer<String, String>(props);
        topic = topicName;
        isAsync = asyncEnable;
        this.mineCode = mineCode;
    }
    @Override
    public void run() {
        // 构造消息记录
        while (true) {
            StringBuffer message = this.getZCSSMessage(this.mineCode);
            long startTime = System.currentTimeMillis();
            String key = String.valueOf(mineCode);
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, message.toString());
            if (isAsync) {
                // 异步发送
                producer.send(record);
            } else {
                try {
                    // 同步发送
                    producer.send(record).get();
                    System.out.println("发送成功:"+message);
                } catch (InterruptedException ie) {
                    LOG.info("The InterruptedException occured : {}.", ie);
                } catch (ExecutionException ee) {
                    LOG.info("The ExecutionException occured : {}.", ee);
                }
            }
            try {
                // 线程阻塞 1 分钟 (60000 毫秒)
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
    private StringBuffer getZCSSMessage(String mineCode) {
        String mineName = KafkaProperties.getMineNameByMineCode(mineCode);
        String systemName = "综采系统";
        String systemCode = "ZCSS";
        String statusCode="33007";  //测点类型编码-运行状态
        String kg = "KG";
        String deviceCode1 = "000000000001";
        String deviceCode2 = "000000000002";
        String deviceCode3 = "000000000003";
        String deviceCode4 = "000000000004";
        String deviceCode5 = "000000000005";
        String deviceCode51 = "000000000015";
        String deviceCode52 = "000000000025";
        String deviceCode53 = "000000000035";
        String deviceCode54 = "000000000045";
        String deviceCode55 = "000000000055";
        String deviceCode56 = "000000000065";
        String deviceCode57 = "000000000075";
        String deviceCode58 = "000000000085";

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        //构建消息内容
        StringBuffer message = new StringBuffer();
        //综采系统接入频率：每60s上传一次测点实时数据。
        //数据上传时间
        LocalDateTime now = LocalDateTime.now();
        // Define the desired format
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // Format the current date and time
        //数据上传时间
        String formattedDateTime = now.format(formatter);


        //构建消息头：包括煤矿编码，煤矿名称，系统型号，系统名称，数据上传时间
        String messageHeader = mineCode+";"+mineName+";;"+systemName+";" + formattedDateTime+"~";
        message.append(messageHeader);



        //（1）采煤机运行状态
        //（2）转载机运行状态
        //（3）前刮板机运行状态
        //（4）1#皮带运行状态，2#皮带运行状态（多个顺槽皮带，序号依次递增）
        //（5）后刮板机运行状态
        StringBuffer messageBody = new StringBuffer();
        //构建消息体：测点编码，测点名称，测点数值，测点数值单位，测点状态，数据时间~



        //获取 综采数据
        SystemService systemService = SpringContextUtil.getBean(SystemService.class);
        Map<String, SynergiaNowPo> nowPoMap = systemService.getZCSystem(mineCode);

        //获取煤流数据
        Map<String, CoalflowNowPo> pdSSystemMap = systemService.getPdSSystemStatus(mineCode);
        if(nowPoMap!=null && nowPoMap.size()>0){
            //获取采煤机运行状态
            String zcStart1="";
            //转载机运行状态
            String zcStartZhuanzai="";
            //前刮板机运行状态
            String zcStartFront="";
            //1#皮带运行状态，2#皮带运行状态（多个顺槽皮带，序号依次递增）
            String zcStartBelt="";
            String zcStartBelt1="";
            String zcStartBelt2="";
            String zcStartBelt3="";
            String zcStartBelt4="";
            String zcStartBelt5="";
            String zcStartBelt6="";
            String zcStartBelt7="";
            String zcStartBelt8="";
            //（5）后刮板机运行状态
            String zcStartBack="";
            if ("150781013671".equals(mineCode)) {
                //灵露
                zcStart1 = "NewSite_采煤_煤机方向";
            }else if ("620824009734".equals(mineCode)) {
                //研北
                zcStart1 = "NewSite_XTKC_采煤机_上电状态";
                zcStartZhuanzai="NewSite_XTKC_转载机_运行状态";
                zcStartFront="NewSite_XTKC_前运机_运行状态";
                zcStartBack="NewSite_XTKC_后运机_运行状态";
                zcStartBelt="NewSite_XTKC_皮带_运行状态";
            }else if ("150621011015".equals(mineCode)) {
                //高头窑
                zcStart1 = "ST26_CMJ_BL1";
                zcStartZhuanzai="ST26_ZZJ_YX";
                zcStartFront="ST26_QBYMK_YX";
                zcStartBack="ST26_HBYMJ_YX";
                //主运
                zcStartBelt="ST01_DJ1_YX";
                //西郊二
                zcStartBelt1="ST01_XYEB_XYEB_YXZT";
                //西郊一
                zcStartBelt2="ST01_XJPLC_HNYX";
                //东郊
                zcStartBelt3="ST01_HNDJ_15";
                //101
                zcStartBelt4="ST01_HNKT101_38";
                //109
                zcStartBelt5="ST01_107PD_YX";
                //转载
                zcStartBelt6="ST01_XJPLC_ZZ_HNYX";
                //102
                zcStartBelt7="ST01_GBJ_102_YX";
                //103
                zcStartBelt8="ST01_GBJ_103_YX";
            }
            SynergiaNowPo zcSystemStatus = nowPoMap.get(zcStart1);
            SynergiaNowPo zcSystemStatusZhuanzai = nowPoMap.get(zcStartZhuanzai);
            SynergiaNowPo zcSystemStatusFront = nowPoMap.get(zcStartFront);
            SynergiaNowPo zcSystemStatusBack = nowPoMap.get(zcStartBack);
            SynergiaNowPo zcSystemStatusBelt = null;
            if ("150621011015".equals(mineCode)) {
                if (pdSSystemMap != null && pdSSystemMap.size() > 0) {
                    CoalflowNowPo coalflowNowPo = pdSSystemMap.get(zcStartBelt);
                    zcSystemStatusBelt = new SynergiaNowPo();
                    zcSystemStatusBelt.setPointValue(coalflowNowPo.getPointValue() != null ? coalflowNowPo.getPointValue() : 0);
                    zcSystemStatusBelt.setDataTime(coalflowNowPo.getDataTime() != null ? coalflowNowPo.getDataTime() : new Date());
                    zcSystemStatusBelt.setPointStatus(coalflowNowPo.getPointStatus()!= null ? coalflowNowPo.getPointStatus() : 1);
                }
            } else {
                 zcSystemStatusBelt = nowPoMap.get(zcStartBelt);
            }

            if (zcSystemStatus != null) {
                SynergiaNowPo zcVo1 = new SynergiaNowPo();
                zcVo1.setDataTime(zcSystemStatus.getDataTime() != null ? zcSystemStatus.getDataTime() : new Date());
                zcVo1.setPointValue(Float.valueOf(zcSystemStatus.getPointValue() != null && zcSystemStatus.getPointValue() > 0 ? 1 : 0));
                zcVo1.setPointStatus(zcSystemStatus.getPointStatus() != null ? zcSystemStatus.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode1+";采煤机运行状态;"+zcVo1.getPointValue().intValue()+";;"+zcVo1.getPointStatus()+";"+simpleDateFormat.format(zcVo1.getDataTime())+"~");
            }
            if (zcSystemStatusZhuanzai != null) {
                SynergiaNowPo zhuanzai = new SynergiaNowPo();
                zhuanzai.setDataTime(zcSystemStatusZhuanzai.getDataTime() != null ? zcSystemStatusZhuanzai.getDataTime() : new Date());
                zhuanzai.setPointValue(Float.valueOf(zcSystemStatusZhuanzai.getPointValue() != null && zcSystemStatusZhuanzai.getPointValue() > 0 ? 1 : 0));
                zhuanzai.setPointStatus(zcSystemStatusZhuanzai.getPointStatus() != null ? zcSystemStatusZhuanzai.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode2+";转载机运行状态;"+zhuanzai.getPointValue().intValue()+";;"+zhuanzai.getPointStatus()+";"+simpleDateFormat.format(zhuanzai.getDataTime())+"~");
            }
            if (zcSystemStatusFront != null) {
                SynergiaNowPo front = new SynergiaNowPo();
                front.setDataTime(zcSystemStatusFront.getDataTime() != null ? zcSystemStatusFront.getDataTime() : new Date());
                front.setPointValue(Float.valueOf(zcSystemStatusFront.getPointValue() != null && zcSystemStatusFront.getPointValue() > 0 ? 1 : 0));
                front.setPointStatus(zcSystemStatusFront.getPointStatus() != null ? zcSystemStatusFront.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode3+";前刮板机运行状态;"+front.getPointValue().intValue()+";;"+front.getPointStatus()+";"+simpleDateFormat.format(front.getDataTime())+"~");
            }
            if (zcSystemStatusBack != null) {
                SynergiaNowPo back = new SynergiaNowPo();
                back.setDataTime(zcSystemStatusBack.getDataTime() != null ? zcSystemStatusBack.getDataTime() : new Date());
                back.setPointValue(Float.valueOf(zcSystemStatusBack.getPointValue() != null && zcSystemStatusBack.getPointValue() > 0 ? 1 : 0));
                back.setPointStatus(zcSystemStatusBack.getPointStatus() != null ? zcSystemStatusBack.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode4+";后刮板机运行状态;"+back.getPointValue().intValue()+";;"+back.getPointStatus()+";"+simpleDateFormat.format(back.getDataTime())+"~");
            }
            if (zcSystemStatusBelt != null) {
                SynergiaNowPo belt = new SynergiaNowPo();
                belt.setDataTime(zcSystemStatusBelt.getDataTime() != null ? zcSystemStatusBelt.getDataTime() : new Date());
                belt.setPointValue(Float.valueOf(zcSystemStatusBelt.getPointValue() != null && zcSystemStatusBelt.getPointValue() > 0 ? 1 : 0));
                belt.setPointStatus(zcSystemStatusBelt.getPointStatus() != null ? zcSystemStatusBelt.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode5+";1#皮带运行状态;"+belt.getPointValue().intValue()+";;"+belt.getPointStatus()+";"+simpleDateFormat.format(belt.getDataTime())+"~");
            }
            //皮带集合
            if (pdSSystemMap != null && pdSSystemMap.size() > 0) {
                CoalflowNowPo coalflowNowPo1 = pdSSystemMap.get(zcStartBelt1);
                CoalflowNowPo coalflowNowPo2 = pdSSystemMap.get(zcStartBelt2);
                CoalflowNowPo coalflowNowPo3 = pdSSystemMap.get(zcStartBelt3);
                CoalflowNowPo coalflowNowPo4 = pdSSystemMap.get(zcStartBelt4);
                CoalflowNowPo coalflowNowPo5 = pdSSystemMap.get(zcStartBelt5);
                CoalflowNowPo coalflowNowPo6 = pdSSystemMap.get(zcStartBelt6);
                CoalflowNowPo coalflowNowPo7 = pdSSystemMap.get(zcStartBelt7);
                CoalflowNowPo coalflowNowPo8 = pdSSystemMap.get(zcStartBelt8);
                if (coalflowNowPo1 != null) {
                    SynergiaNowPo belt1 = new SynergiaNowPo();
                    belt1.setDataTime(coalflowNowPo1.getDataTime() != null ? coalflowNowPo1.getDataTime() : new Date());
                    belt1.setPointValue(Float.valueOf(coalflowNowPo1.getPointValue() != null && coalflowNowPo1.getPointValue() > 0 ? 1 : 0));
                    belt1.setPointStatus(coalflowNowPo1.getPointStatus() != null ? coalflowNowPo1.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode51+";2#皮带运行状态;"+belt1.getPointValue().intValue()+";;"+belt1.getPointStatus()+";"+simpleDateFormat.format(belt1.getDataTime())+"~");
                }
                if (coalflowNowPo2 != null) {
                    SynergiaNowPo belt2 = new SynergiaNowPo();
                    belt2.setDataTime(coalflowNowPo2.getDataTime() != null ? coalflowNowPo2.getDataTime() : new Date());
                    belt2.setPointValue(Float.valueOf(coalflowNowPo2.getPointValue() != null && coalflowNowPo2.getPointValue() > 0 ? 1 : 0));
                    belt2.setPointStatus(coalflowNowPo2.getPointStatus() != null ? coalflowNowPo2.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode52+";3#皮带运行状态;"+belt2.getPointValue().intValue()+";;"+belt2.getPointStatus()+";"+simpleDateFormat.format(belt2.getDataTime())+"~");
                }
                if (coalflowNowPo3 != null) {
                    SynergiaNowPo belt3 = new SynergiaNowPo();
                    belt3.setDataTime(coalflowNowPo3.getDataTime() != null ? coalflowNowPo3.getDataTime() : new Date());
                    belt3.setPointValue(Float.valueOf(coalflowNowPo3.getPointValue() != null && coalflowNowPo3.getPointValue() > 0 ? 1 : 0));
                    belt3.setPointStatus(coalflowNowPo3.getPointStatus() != null ? coalflowNowPo3.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode53+";4#皮带运行状态;"+belt3.getPointValue().intValue()+";;"+belt3.getPointStatus()+";"+simpleDateFormat.format(belt3.getDataTime())+"~");
                }
                if (coalflowNowPo4 != null) {
                    SynergiaNowPo belt4 = new SynergiaNowPo();
                    belt4.setDataTime(coalflowNowPo4.getDataTime() != null ? coalflowNowPo4.getDataTime() : new Date());
                    belt4.setPointValue(Float.valueOf(coalflowNowPo4.getPointValue() != null && coalflowNowPo4.getPointValue() > 0 ? 1 : 0));
                    belt4.setPointStatus(coalflowNowPo4.getPointStatus() != null ? coalflowNowPo4.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode54+";5#皮带运行状态;"+belt4.getPointValue().intValue()+";;"+belt4.getPointStatus()+";"+simpleDateFormat.format(belt4.getDataTime())+"~");
                }
                if (coalflowNowPo5 != null) {
                    SynergiaNowPo belt5 = new SynergiaNowPo();
                    belt5.setDataTime(coalflowNowPo5.getDataTime() != null ? coalflowNowPo5.getDataTime() : new Date());
                    belt5.setPointValue(Float.valueOf(coalflowNowPo5.getPointValue() != null && coalflowNowPo5.getPointValue() > 0 ? 1 : 0));
                    belt5.setPointStatus(coalflowNowPo5.getPointStatus() != null ? coalflowNowPo5.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode55+";6#皮带运行状态;"+belt5.getPointValue().intValue()+";;"+belt5.getPointStatus()+";"+simpleDateFormat.format(belt5.getDataTime())+"~");
                }
                if (coalflowNowPo6 != null) {
                    SynergiaNowPo belt6 = new SynergiaNowPo();
                    belt6.setDataTime(coalflowNowPo6.getDataTime() != null ? coalflowNowPo6.getDataTime() : new Date());
                    belt6.setPointValue(Float.valueOf(coalflowNowPo6.getPointValue() != null && coalflowNowPo6.getPointValue() > 0 ? 1 : 0));
                    belt6.setPointStatus(coalflowNowPo6.getPointStatus() != null ? coalflowNowPo6.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode56+";7#皮带运行状态;"+belt6.getPointValue().intValue()+";;"+belt6.getPointStatus()+";"+simpleDateFormat.format(belt6.getDataTime())+"~");
                }
                if (coalflowNowPo7 != null) {
                    SynergiaNowPo belt7 = new SynergiaNowPo();
                    belt7.setDataTime(coalflowNowPo7.getDataTime() != null ? coalflowNowPo7.getDataTime() : new Date());
                    belt7.setPointValue(Float.valueOf(coalflowNowPo7.getPointValue() != null && coalflowNowPo7.getPointValue() > 0 ? 1 : 0));
                    belt7.setPointStatus(coalflowNowPo7.getPointStatus() != null ? coalflowNowPo7.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode57+";8#皮带运行状态;"+belt7.getPointValue().intValue()+";;"+belt7.getPointStatus()+";"+simpleDateFormat.format(belt7.getDataTime())+"~");
                }
                if (coalflowNowPo8 != null) {
                    SynergiaNowPo belt8 = new SynergiaNowPo();
                    belt8.setDataTime(coalflowNowPo8.getDataTime() != null ? coalflowNowPo8.getDataTime() : new Date());
                    belt8.setPointValue(Float.valueOf(coalflowNowPo8.getPointValue() != null && coalflowNowPo8.getPointValue() > 0 ? 1 : 0));
                    belt8.setPointStatus(coalflowNowPo8.getPointStatus() != null ? coalflowNowPo8.getPointStatus() : 1);
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode58+";9#皮带运行状态;"+belt8.getPointValue().intValue()+";;"+belt8.getPointStatus()+";"+simpleDateFormat.format(belt8.getDataTime())+"~");
                }

            }
        }


        //构建消息结尾
        messageBody.append("||");
        message.append(messageBody);
        return message;
    }
}
